package com.example.mq.sender;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.example.mq.dto.Msg;
import com.example.mq.enums.MsgStatusEnum;
import com.example.mq.lock.DistributeLock;
import com.example.mq.po.MsgDelayPO;
import com.example.mq.po.MsgPO;
import com.example.mq.retry.MqSendRetry;
import com.example.mq.retry.MqSendRetryResult;
import com.example.mq.service.ISequentialMsgNumberGeneratorService;
import com.example.mq.service.MsgDelayService;
import com.example.mq.service.MsgService;
import com.example.utils.CollUtils;
import com.example.utils.DelayTaskProcessorUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class DefaultMsgSender implements IMsgSender
{
    private MsgService msgService;
    private MqSendRetry mqSendRetry;
    private ThreadPoolTaskExecutor mqExecutor;
    // 分布式锁
    private DistributeLock distributedLock;
    // 延迟消息服务
    private MsgDelayService mqDelayService;

    private RabbitTemplate rabbitTemplate;

    private TransactionTemplate transactionTemplate;

    private ISequentialMsgNumberGeneratorService sequentialMsgNumberGeneratorService;


    // 在延迟队列（delayMsgProcessor）等待被投递的延迟消息
    private Map<String, MsgDelayPO> delayMsgWaitingMap = new ConcurrentHashMap<>();

    private DelayTaskProcessorUtils delayMsgProcessor = new DelayTaskProcessorUtils("delayMsgProcessor", 1, 4);


    // 在延迟队列（sendRetryProcessor）等待被投递重试的消息
    private Map<String, MsgDelayPO> delaySendRetryWaitingMap = new ConcurrentHashMap<>();
    // 延迟队列：处理投递失败后需要延迟投递重试的消息
    private DelayTaskProcessorUtils delaySendRetryProcessor = new DelayTaskProcessorUtils("delaySendRetryProcessor", 1, 4);

    public DefaultMsgSender(MsgService msgService,
                            MqSendRetry mqSendRetry,
                            ThreadPoolTaskExecutor mqExecutor,
                            DistributeLock distributedLock,
                            MsgDelayService mqDelayService,
                            RabbitTemplate rabbitTemplate,
                            TransactionTemplate transactionTemplate
    )
    {
        this.msgService = msgService;
        this.mqSendRetry = mqSendRetry;
        this.mqExecutor = mqExecutor;
        this.distributedLock = distributedLock;
        this.mqDelayService = mqDelayService;
        this.rabbitTemplate =  rabbitTemplate;
        this.transactionTemplate = transactionTemplate;
    }

    /*
     * @param msgList
     * @==============History===============<br/>;
     * @Description //   批量发送消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    @Override
    public void send(List<Object> msgList)
    {
        //将消息转换为json格式
        List<String> msgBodyJsonList = CollUtils.convertList(msgList, JSONUtil::toJsonStr);

        //当前是否有事务？
        boolean hasTransaction = hasTransaction();
        if (hasTransaction)
        {
            // 有事务，则先入库
            List<MsgPO> msgPOList = this.msgService.batchInsert(msgBodyJsonList);
            /*
             * spring事务扩展点，通过TransactionSynchronizationManager.registerSynchronization添加一个事务同步器TransactionSynchronization，
             * 事务执行完成之后，不管事务成功还是失败，都会调用TransactionSynchronization#afterCompletion 方法
             */
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization()
            {
                @Override
                public void afterCompletion(int status)
                {
                    // 为了提升性能：事务消息的投递消息这里异步去执行，即使失败了，会有补偿JOB进行重试
                    mqExecutor.execute(() -> transactionAfter(msgPOList));
                }
            });
        }
        else
        {
            // 没有事务，则直接投递消息到MQ
            for (String msgBodyJson : msgBodyJsonList)
            {
                this.sendMsgToMQ(msgBodyJson);
            }
        }
    }



    /*
     * @param msgPO
     * @==============History===============<br/>;
     * @Description //   投递重试（针对于立刻发送到mq的消息）
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    @Override
    public void sendRetry(MsgPO msgPO)
    {
        this.sendMsg(msgPO.getId());
    }

    @Override
    public <T> Msg<T> build(T msgBody)
    {
        return Msg.msg("lesson", IdUtil.fastSimpleUUID(), msgBody);
    }

    @Override
    public void sendSequentialMsg(String busId, String exchange, String routingKey, Msg<?> msg)
    {
        Objects.nonNull(busId);

        String groupId = String.format("busId:%s,exchange:%s,routingKey:%s", busId, StringUtils.defaultString(exchange), StringUtils.defaultString(routingKey));

        //加分布式锁，保证同一个groupId时，顺序消息发送串行执行
        this.distributedLock.accept("sendSequentialMsg:" + groupId, lockKey ->
        {
            this.transactionTemplate.executeWithoutResult(action ->
            {
                //设置顺序消息groupId
                msg.setSequentialMsgGroupId(groupId);
                //设置顺序消息的编号
                msg.setSequentialMsgNumbering(this.sequentialMsgNumberGeneratorService.get(groupId));
                this.send(msg);
            });
        });
    }


    /*
     * 事务后执行的方法
     * @param msgPOList
     */
    private void transactionAfter(List<MsgPO> msgPOList)
    {
    if (null != msgPOList  && msgPOList.size() > 0)
        {
            // 循环发送消息
            for (MsgPO msg : msgPOList)
            {
                this.sendMsg(msg.getId());
            }
            log.info("事务执行成功，消息投递完毕");
        } else
        {
            // 走到这里，说明事务执行失败了，消息未投递，由于插入消息和业务在一个事务中，事务执行失败，此时db中也是没有消息记录
            log.info("事务执行失败，消息未投递");
        }
    }

    /*
     * @param msgPO
     * @==============History===============<br/>;
     * @Description //   立刻投递MQ消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void sendMsg(String msgId)
    {
        distributedLock.accept("sendMsg:" + msgId,
                lockKey -> {
                    // 加锁成功后，重新从db中获取消息
                    MsgPO msgPO = this.msgService.getById(msgId);
                    //若消息已投递成功 || （投递失败且不需要重试了），则不处理
                    if (MsgStatusEnum.SUCCESS.getStatus().equals(msgPO.getStatus()) ||
                            (MsgStatusEnum.FAIL.getStatus().equals(msgPO.getStatus()) && msgPO.getSendRetry().equals(0)))
                    {
                        return;
                    }

                    Exception exception = null;
                    try
                    {
                        //投递消息到MQ
                        sendMsgToMQ(msgPO.getBodyJson());
                    }
                    catch (Exception e)
                    {
                        exception = e;
                    }
                    //没有异常，说明消息投递成功
                    if (exception == null)
                    {
                        //将消息状态置为成功
                        msgService.updateStatusSuccess(msgPO);
                    } else
                    {
                        // 有异常，则获取重试信息，将信息更新到db中
                        MqSendRetryResult retryResult = this.mqSendRetry.getRetryResult(msgPO);
                        //异常信息
                        String failMsg = ExceptionUtil.stacktraceToString(exception);
                        //是否需要重试？
                        boolean retry = retryResult.isRetry();
                        //下次重试时间
                        LocalDateTime nextRetryTime = retryResult.getNextRetryTime();
                        //将消息状态置为失败（且记录失败信息、是否需要重试，下次重试时间）
                        this.msgService.updateStatusFail(msgPO, failMsg, retry, nextRetryTime);

                        //如果走到这里，还是失败，且不需要重试了，建议告警
                        if (!retry)
                        {
                            // todo 请执行告警代码，人工干预
                        }
                    }
                });
    }







    /*
     * @param msgList
     * @==============History===============<br/>;
     * @Description //   批量投递延迟消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    @Override
    public void send(long delayTime, TimeUnit delayTimeUnit, List<Object> msgList)
    {
        if (delayTime <= 0)
        {
            throw new IllegalArgumentException("Invalid parameter: delayTimeUnit, must be greater than 0");
        }
        Objects.nonNull(delayTimeUnit);

        // 计算期望发送时间
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime expectSendTime = now.plusSeconds(TimeUnit.SECONDS.convert(delayTime, delayTimeUnit));

        // 发送消息
        this.send(expectSendTime, msgList);
    }



    /*
     * @param expectSendTime
     * @param msgList
     * @==============History===============<br/>;
     * @Description //   向延时队列中丢任务
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void send(LocalDateTime expectSendTime, List<Object> msgList)
    {
        Objects.nonNull(expectSendTime);
        // 将消息转换为json格式
        List<String> msgBodyJsonList = CollUtils.convertList(msgList, JSONUtil::toJsonStr);

        // 消息是否需要先入库？（事务消息 || 延迟消息）需要先入库；延迟消息为什么需要入库？有可能延迟10天，那么这个数据会先存储到db，会依赖数据库
        if (this.isNeedStoreToDb(expectSendTime))
        {
            // 有事务，则先入库
            List<MsgDelayPO> msgPOList = this.mqDelayService.batchInsert(expectSendTime, msgBodyJsonList);
            boolean hasTransaction = hasTransaction();
            if (!hasTransaction)
            {
                mqExecutor.execute(() -> deliverMsg(msgPOList));
            }else
            {
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization()
                {
                    @Override
                    public void afterCompletion(int status)
                    {
                        /**
                         * 代码走到这里时，事务已经完成了（可能是回滚了、或者是提交了）
                         * 看下本地消息记录是否存在？如果存在，说明事务是成功的，业务是执行成功的，则投递消息 & 并将消息状态置为成功
                         */
                        MsgDelayPO msgDelayPO =  mqDelayService.getById(msgPOList.get(0).getId());
                        if (msgDelayPO == null)
                        {
                            return;
                        }
                        //为了提升性能：事务消息的投递消息这里异步去执行，即使失败了，会有补偿JOB进行重试
                        mqExecutor.execute(() -> deliverMsg(msgPOList));
                    }
                });
            }
        }
        else
        {
            // 不需要入库的，直接投递
            for (String msgBodyJson : msgBodyJsonList)
            {
                this.sendMsgToMQ(msgBodyJson);
            }
        }
    }


    /*
     * @param msgPO
     * @==============History===============<br/>;
     * @Description //  投递重试（针对于延迟发送到mq的消息）
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    @Override
    public void sendDelayRetry(MsgDelayPO msgDelayPO)
    {
        if (msgDelayPO == null)
            return;
        if (MsgStatusEnum.INIT.getStatus().equals(msgDelayPO.getStatus()))
        {
            // 待投递的
            this.deliverMsg(msgDelayPO);
        }
        else if (MsgStatusEnum.FAIL.getStatus().equals(msgDelayPO.getStatus()) && msgDelayPO.getSendRetry().equals(1))
        {
            // 投递失败的，需要重试的
            this.delaySendRetry(msgDelayPO);
        }
    }

    /*
     * @param
     * @param msgPO
     * @==============History===============<br/>;
     * @Description //   处理投递失败的，需要重试的的延时消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void delaySendRetry(MsgDelayPO msgDelayPO)
    {
        // 失败了，还需要重试，则丢入延迟队列（delayRetryProcessor），稍后继续重试
        // 计算延迟时间（s）
        LocalDateTime now = LocalDateTime.now();
        long delayTimeMills = ChronoUnit.MILLIS.between(now, msgDelayPO.getNextRetryTime());
        if (delayTimeMills <= TimeUnit.MINUTES.toMillis(2))
        {
            // 为了防止消息重复排队，这里使用了一个map，map中没有的时候，才放入队列进行排队，从队列中移除的时候，会从map中移除
            delaySendRetryWaitingMap.computeIfAbsent(msgDelayPO.getId(), key ->
            {
                boolean putSuccess = delaySendRetryProcessor.put(delayTimeMills, TimeUnit.MILLISECONDS, () ->
                {
                    delaySendRetryWaitingMap.remove(msgDelayPO.getId());
                    deliverMsg(mqDelayService.getById(msgDelayPO.getId()));
                });
                if (!putSuccess)
                {
                    log.error("本地延迟投递重试队列已满：{}", this.delaySendRetryProcessor);
                    throw new RuntimeException("本地延迟投递重试队列已满，延迟投递重试排队失败");
                }
                return msgDelayPO;
            });
        }
    }


    /*
     * @param list
     * @==============History===============<br/>;
     * @Description //   批量投递延迟消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void deliverMsg(List<MsgDelayPO> list)
    {
        for (MsgDelayPO msgDelayPO : list)
        {
            deliverMsg(msgDelayPO);
        }
    }

    /*
     * @param
     * @param msgDelayPO
     * @==============History===============<br/>;
     * @Description //   处理待投递的延时消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void deliverMsg(MsgDelayPO msgDelayPO)
    {
        //如果期望发送时间>当前时间，则走延迟消息发送处理逻辑
        if (msgDelayPO.getExpectSendTime().isAfter(LocalDateTime.now()))
        {
            // 延迟消息的投递
            this.deliverDelayMsg(msgDelayPO);
        }
        else
        {
            // 投递消息
            this.deliverImmediateMsg(msgDelayPO.getId());
        }
    }

    /*
     * @param id
     * @==============History===============<br/>;
     * @Description // 立刻投递消息
     * @Date 2025/8/14 
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void deliverImmediateMsg(String id)
    {
      // 对消息添加分布式锁，防止服务集群部署时，避免重复投递
        distributedLock.accept("deliverImmediateMsg:" + id,
                lockKey -> {
                    //加锁成功后，重新从db中获取消息
                    MsgDelayPO msgDelayPO = mqDelayService.getById(id);
                    //若消息已投递成功 || （投递失败且不需要重试了），则不处理
                    if (MsgStatusEnum.SUCCESS.getStatus().equals(msgDelayPO.getStatus()) ||
                            (MsgStatusEnum.FAIL.getStatus().equals(msgDelayPO.getStatus()) && msgDelayPO.getSendRetry().equals(0)))
                    {
                        return;
                    }
                    Exception exception = null;
                    try
                    {
                        //投递消息到MQ
                        sendMsgToMQ(msgDelayPO.getBodyJson());
                    } catch (Exception e)
                    {
                        exception = e;
                    }
                    //没有异常，说明消息投递成功
                    if (exception == null)
                    {
                        //将消息状态置为成功
                        mqDelayService.updateStatusSuccess(msgDelayPO);
                    } else
                    {
                        //有异常，则获取重试信息，将信息更新到db中
                        MqSendRetryResult retryResult = mqSendRetry.getDelayRetryResult(msgDelayPO);
                        //异常信息
                        String failMsg = ExceptionUtil.stacktraceToString(exception);
                        //是否需要重试？
                        boolean retry = retryResult.isRetry();
                        //下次重试时间
                        LocalDateTime nextRetryTime = retryResult.getNextRetryTime();
                        //将消息状态置为失败（且记录失败信息、是否需要重试，下次重试时间）
                        mqDelayService.updateStatusFail(msgDelayPO, failMsg, retry, nextRetryTime);

                        if (retry)
                        {
                            //投递失败，需要重试的，则延迟重试
                            this.delaySendRetry(mqDelayService.getById(msgDelayPO.getId()));
                        } else
                        {
                            //如果走到这里，还是失败，且不需要重试了，建议告警，说明重试次数达到上限了，请求人工干预
                            //todo 请执行告警代码，人工干预
                        }
                    }
                });
    }

    /*
     * @param
     * @param msgDelayPO
     * @==============History===============<br/>;
     * @Description //   投递延时消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void deliverDelayMsg(MsgDelayPO msgDelayPO)
    {
        // 计算延迟时间（s）
        LocalDateTime now = LocalDateTime.now();
        long delayTimeMills = ChronoUnit.MILLIS.between(now, msgDelayPO.getExpectSendTime());

        // 这里会将2分中内待发送的消息，放到java本地延迟队列排队处理，非2分钟内的消息，交给job去触发
        if (delayTimeMills <= TimeUnit.MINUTES.toMillis(2))
        {
            // 为了防止消息重复排队，这里使用了一个map，map中没有的时候，才放入队列进行排队，从队列中移除的时候，会从map中移除
            delayMsgWaitingMap.computeIfAbsent(msgDelayPO.getId(), msgId ->
            {
                boolean putSuccess = delayMsgProcessor.put(delayTimeMills, TimeUnit.MILLISECONDS, () ->
                {
                    // 从map中移除
                    this.delayMsgWaitingMap.remove(msgId);
                    //立即投递消息
                    this.deliverImmediateMsg(msgDelayPO.getId());
                });
                if (!putSuccess)
                {
                    log.error("本地延迟队列已满：{}", this.delayMsgProcessor);
                    throw new RuntimeException("本地延迟队列已满,延迟消息排队失败");
                }
                return msgDelayPO;
            });
        }

    }


    /*
     * 判断当前是否有事务？
     * @return
     */
    private static boolean hasTransaction()
    {
        //开启了事务同步 && 并且当前确实有事务
        return TransactionSynchronizationManager.isSynchronizationActive() && TransactionSynchronizationManager.isActualTransactionActive();
    }



    /*
     * @param msgBodyJson
     * @==============History===============<br/>;
     * @Description //   向MQ发送消息
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void sendMsgToMQ(String msgBodyJson)
    {
        log.info("开始投递消息到MQ。。。。");
        log.info("消息体：{}", msgBodyJson);
        // 这里为了测试出投递失败的效果，故意在这里加了一段代码
//        if (msgBodyJson.length() > 100)
//        {
//            throw new RuntimeException("模拟投递失败的情况，故意失败：" + msgBodyJson);
//        }

        this.rabbitTemplate.convertAndSend("orderExchange", "create", msgBodyJson);
        log.info("消息已投递到MQ");
    }

    /*
     * @param expectSendTime
     * @param msgList
     * @==============History===============<br/>;
     * @Description //   消息是否需要先存储到db
     * @Date 2025/8/14 
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private boolean isNeedStoreToDb(LocalDateTime expectSendTime)
    {
        //事务消息 || 延迟消息
        return hasTransaction() || expectSendTime.isAfter(LocalDateTime.now());
    }

}
